﻿//------------------------------------------------------------
// by mky 2022/6/25
//------------------------------------------------------------
#include "ftmqtt.h"
#include <chrono>
#include <functional>
#include <unistd.h>
#include <iostream>
#include "common/utils/SimpleIni.h"
#include "common/version.hpp"
#include "Json\value.h"
/* Include the autoconf generated config.h */
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include "wolfmqtt/mqtt_client.h"

/* 此示例仅适用于ENABLE\u MQTT\u TLS（wolfSSL库）* /
/* 注意：
*
*“wc_GetTime”API是在3.9.1中添加的，如果没有，则需要实现
* 您自己的版本，以获取当前UTC秒数或更新您的wolfSSL库
*/

/*此示例需要wolfSSL 3.9.1或更高版本中的功能*/

#include <wolfssl/wolfcrypt/settings.h>
#include <wolfssl/version.h>

#include <wolfssl/ssl.h>
#include <wolfssl/wolfcrypt/asn_public.h>
#include <wolfssl/wolfcrypt/coding.h>
#include <wolfssl/wolfcrypt/hmac.h>

#include "common/mqtt/mqttexample.h"
#include "common/mqtt/mqttnet.h"
#include <wolfmqtt/version.h>

static wm_Sem mtLock; /* Protect "packetId" and "stop" */
#define TEST_MESSAGE    "test00"
//static int mStopRead = 0;
static int mNumMsgsRecvd;
static int mNumMsgsDone;

/* Locals */
static int mStopRead = 0;
static bool keep = false;
/* Configuration */
#define APP_HARDWARE         "wolf_aws_iot_demo"
#define APP_FIRMWARE_VERSION LIBWOLFMQTT_VERSION_STRING

#define MAX_BUFFER_SIZE         10240    /* Maximum size for network read/write callbacks */
#define AWSIOT_HOST             "192.168.103.220"
//#define AWSIOT_PORT             "8883"
#define AWSIOT_DEVICE_ID        "emax_test_2"
#define AWSIOT_QOS              MQTT_QOS_1
#define AWSIOT_KEEP_ALIVE_SEC   DEFAULT_KEEP_ALIVE_SEC
#define AWSIOT_CMD_TIMEOUT_MS   DEFAULT_CMD_TIMEOUT_MS

#define AWSIOT_SUBSCRIBE_TOPIC  "$aws/things/" AWSIOT_DEVICE_ID "/shadow/update"
#define AWSIOT_PUBLISH_TOPIC    "$aws/things/" AWSIOT_DEVICE_ID "/shadow/update"

#define AWSIOT_PUBLISH_MSG_SZ   400
#define AWSIOT_USERNAME         "emqx_test"
#define AWSIOT_PASSWORD         "anb*&nnm2266"
static const char* root_ca =
"-----BEGIN CERTIFICATE-----\n"
"MIIDhjCCAm6gAwIBAgIUaBoiEk9dTtvUJORnHm5Fw0As3howDQYJKoZIhvcNAQEL\n"
"BQAwSTELMAkGA1UEBhMCQ04xEDAOBgNVBAgTB0JlaWppbmcxEDAOBgNVBAcTB0Jl\n"
"aWppbmcxFjAUBgNVBAMTDW1heHZpc2lvbnRlc3QwHhcNMjIwODA1MDg1ODAwWhcN\n"
"MjcwODA0MDg1ODAwWjBJMQswCQYDVQQGEwJDTjEQMA4GA1UECBMHQmVpamluZzEQ\n"
"MA4GA1UEBxMHQmVpamluZzEWMBQGA1UEAxMNbWF4dmlzaW9udGVzdDCCASIwDQYJ\n"
"KoZIhvcNAQEBBQADggEPADCCAQoCggEBAMO2g6KG12rnBj2zvRvNUx2VSDwv4oIQ\n"
"eejHiJQEkj7Yv/psGd7a6QX9rg3+8TzVq0tKlckbzcWLGGldagWNayK6X+bXIoUo\n"
"rmGGJFqu+LwR0gQEx6BX3d4B+wAxgEldFYWJMeYZZY4YW7BWNjXMUiDF6Bw2FPh0\n"
"KI8JydcqeNTJ4R4DxcqualcZRXcQVu7qwY/FE07O3+SA51+5Lb+AMbht+TCJfGzj\n"
"SFChz5AtSQwWEX89/zCkh1WxXoUfOBuXlaGfroAe8lqHTbx5nQYpHw4zCaJ2SgiN\n"
"cnYqbAhxq7VVDDU/FqUgfjM8bJp8YyneNUqJ05T3YQBe6PcDD5ceP48CAwEAAaNm\n"
"MGQwDgYDVR0PAQH/BAQDAgEGMBIGA1UdEwEB/wQIMAYBAf8CAQIwHQYDVR0OBBYE\n"
"FJubzqQGKlv7dY2NqOTcVzS8CGVOMB8GA1UdIwQYMBaAFJubzqQGKlv7dY2NqOTc\n"
"VzS8CGVOMA0GCSqGSIb3DQEBCwUAA4IBAQBeSmY6cHiHD/X5xWyk/fh9YcEv6fVr\n"
"eax1fMsyy56z4swYOeSfc65oes7NG+LuPDX5nzhqlo8YTZymFn36XcehMq152/BH\n"
"ZzlnxawfZFnsKvmgPlSlxKDSE4FaAZRyuUxAPt2D5lhjzalEo12W9TnQTAT5LSlD\n"
"cqrtm2sKyS4ZOxHtkzQ9D6cWnRZXpUCLR60T3kSWa7TLt3pia9qAE843EBHOuIut\n"
"LvznBFkHkcmWgm/27Xid8Q6MHDGF45jMAwh5D6/A04XoUYTlILTZZNXREtVJjwT2\n"
"Z5v2UPDJq+l5+/hQO7kUFRgNOhQnQY4UzkwuvAV5ubwStbstVuuKTlLh\n"
"-----END CERTIFICATE-----";

//-------------------------------------------------------------------------------------

typedef struct mv_mqtt_instance_impl_t
{
	std::string broker_address;
	std::string broker_port;
	std::string client_id;
	std::string product_key;
	std::string device_name;
	std::string version = "1.0";
	uint32_t qos;
	uint32_t connection_timeout;
	uint32_t send_timeout;

	std::string trust_store;
	std::string key_store;
	std::string CApath;
	std::string private_key_password;
	std::string sn;
	std::string deviceNum;

	std::string device_secret;
	std::string register_time;
	std::string product_secret;

	std::string username;
	std::string password;

	std::string aes_decrypt_key = "";
	std::string aes_decrypt_iv = "";

	std::string config_path;

	std::string app_id;
	std::string cloud_address;

	std::unordered_map<std::string, std::function<int(void*, const char*, const char*)>> handle_function;

	MQTTCtx mqttCtx;

	void* userdata;
#pragma region CALLBACK
	mqtt_msg_handler_callback_t mqtt_msg_handler_t;
	mqtt_connection_status_callback_t mqtt_conn_status_t;
#pragma endregion

#pragma region TOPIC
	//设备注册
	std::string topic_device_register       = "/maxvision/sys/{productKey}/{deviceName}/maxvision/sys/device/register"; 
	std::string topic_device_register_reply = "/maxvision/sys/{productKey}/{deviceName}/device/register_reply"; 
	//属性设置
	std::string topic_property_set          = "/maxvision/sys/{productKey}/{deviceName}/thing/property/set";
	std::string topic_property_set_reply	= "/maxvision/sys/{productKey}/{deviceName}/thing/property/set_reply";
	//属性获取
	std::string topic_property_get			= "/maxvision/sys/{productKey}/{deviceName}/thing/property/get";
	std::string topic_property_get_reply	= "/maxvision/sys/{productKey}/{deviceName}/thing/property/get_reply";
	//动作
	std::string topic_action_exec			= "/maxvision/sys/{productKey}/{deviceName}/thing/action/{模块id:行为id}/exec";
	std::string topic_action_exec_reply		= "/maxvision/sys/{productKey}/{deviceName}/thing/action/{模块id:行为id}/exec_reply";
	//时序事件
	std::string topic_tsevent_post			= "/maxvision/sys/{productKey}/{deviceName}/thing/tsevent/{模块id:行为id}/post";
	std::string topic_tsevent_post_reply	= "/maxvision/sys/{productKey}/{deviceName}/thing/tsevent/{模块id:行为id}/post_reply";
	//非时序事件
	std::string topic_event_post			= "/maxvision/sys/{productKey}/{deviceName}/thing/event/+/post";
	std::string topic_event_post_reply		= "/maxvision/sys/{productKey}/{deviceName}/thing/event/+/post_reply";
#pragma endregion


} mv_mqtt_instance_impl_t;
bool mv_mqtt_keep() {
	return keep;
}
static word16 mqtt_get_packetid_threadsafe(void)
{
	word16 packet_id;
	wm_SemLock(&mtLock);
	packet_id = mqtt_get_packetid();
	wm_SemUnlock(&mtLock);
	return packet_id;
}
static void mqtt_stop_set(void)
{
	wm_SemLock(&mtLock);
	PRINTF("MQTT Stopping");
	mStopRead = 1;
	wm_SemUnlock(&mtLock);
}

static int mqtt_stop_get(void)
{
	int rc;
	wm_SemLock(&mtLock);
	rc = mStopRead;
	wm_SemUnlock(&mtLock);
	return rc;
}

static int check_response(mv_mqtt_instance_impl_t* impl, int rc, word32* startSec, int packet_type)
{
	/* check for test mode */
	if (mqtt_stop_get()) {
		PRINTF("MQTT Exiting Thread...");
		return MQTT_CODE_SUCCESS;
	}

#ifdef WOLFMQTT_NONBLOCK
#ifdef WOLFMQTT_TEST_CANCEL
	if (packet_type == MQTT_PACKET_TYPE_PUBLISH && rc == MQTT_CODE_CONTINUE) {
		PRINTF("Test cancel by setting early timeout");
		return MQTT_CODE_ERROR_TIMEOUT;
	}
	else
#else
	(void)packet_type;
#endif
	/* Track elapsed time with no activity and trigger timeout */
	rc = mqtt_check_timeout(rc, startSec, impl->mqttCtx.cmd_timeout_ms / 1000);

	/* check return code */
	if (rc == MQTT_CODE_CONTINUE) {
#if 0
		/* optionally add delay when debugging */
		usleep(100 * 1000);
#endif
	}
#else
	(void)packet_type;
	(void)startSec;
	(void)impl->mqttCtx;
#endif
	return rc;
}

int mv_wmq_begin(mv_mqtt_instance_impl_t* impl) {
	PRINTF("AwsIoT Client: QoS %d, Use TLS %d", &impl->mqttCtx.qos, &impl->mqttCtx.use_tls);

	/* Aws IoT requires TLS */
	if (!&impl->mqttCtx.use_tls) {
		return MQTT_CODE_ERROR_BAD_ARG;
	}
	return MQTT_CODE_SUCCESS;
}
void mv_wmq_client_cleanup(mv_mqtt_instance_impl_t* impl) {
	/* Free resources */
	//if (impl->mqttCtx.tx_buf) WOLFMQTT_FREE(impl->mqttCtx.tx_buf);
	//if (impl->mqttCtx.rx_buf) WOLFMQTT_FREE(impl->mqttCtx.rx_buf);

	/* Cleanup network */
	MqttClientNet_DeInit(&impl->mqttCtx.net);

	MqttClient_DeInit(&impl->mqttCtx.client);
}
int mv_wmq_disconn(mv_mqtt_instance_impl_t* impl) {
	int rc;

	do {
		/* Disconnect */
		rc = MqttClient_Disconnect_ex(&impl->mqttCtx.client,
			&impl->mqttCtx.disconnect);
	} while (rc == MQTT_CODE_CONTINUE);

	PRINTF("MQTT Disconnect: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);

	rc = MqttClient_NetDisconnect(& impl->mqttCtx.client);

	PRINTF("MQTT Socket Disconnect: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);

	mv_wmq_client_cleanup(impl);
	//if (impl->mqttCtx.tx_buf) WOLFMQTT_FREE(impl->mqttCtx.tx_buf);
	//if (impl->mqttCtx.rx_buf) WOLFMQTT_FREE(impl->mqttCtx.rx_buf);
	return rc;
}

int mv_wmq_net_init(mv_mqtt_instance_impl_t* impl) {
	/* Initialize Network */
	int rc = MqttClientNet_Init(&impl->mqttCtx.net, &impl->mqttCtx);
	if (rc == MQTT_CODE_CONTINUE) {
		return rc;
	}
	PRINTF("MQTT Net Init: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}
	return MQTT_CODE_SUCCESS;
}
static int mqtt_aws_tls_verify_cb(int preverify, WOLFSSL_X509_STORE_CTX* store)
{
	char buffer[WOLFSSL_MAX_ERROR_SZ];

	PRINTF("MQTT TLS Verify Callback: PreVerify %d, Error %d (%s)", preverify,
		store->error, store->error != 0 ?
		wolfSSL_ERR_error_string(store->error, buffer) : "none");
	PRINTF("  Subject's domain name is %s", store->domain);

	if (store->error != 0) {
		/* Allowing to continue */
		/* Should check certificate and return 0 if not okay */
		PRINTF("  Allowing cert anyways");
	}

	return 1;
}

/* Use this callback to setup TLS certificates and verify callbacks */
static int mqtt_aws_tls_cb(MqttClient* client)
{
	int rc = WOLFSSL_FAILURE;

	/* Use highest available and allow downgrade. If wolfSSL is built with
	 * old TLS support, it is possible for a server to force a downgrade to
	 * an insecure version. */
	client->tls.ctx = wolfSSL_CTX_new(wolfSSLv23_client_method());
	if (client->tls.ctx) {
		wolfSSL_CTX_set_verify(client->tls.ctx, WOLFSSL_VERIFY_PEER,
			mqtt_aws_tls_verify_cb);

		/* Load CA certificate buffer */
		rc = wolfSSL_CTX_load_verify_buffer(client->tls.ctx,
			(const byte*)root_ca, (long)XSTRLEN(root_ca), WOLFSSL_FILETYPE_PEM);

		///* Load Client Cert */
		//if (rc == WOLFSSL_SUCCESS)
		//	rc = wolfSSL_CTX_use_certificate_buffer(client->tls.ctx,
		//		(const byte*)device_cert, (long)XSTRLEN(device_cert),
		//		WOLFSSL_FILETYPE_PEM);

		///* Load Private Key */
		//if (rc == WOLFSSL_SUCCESS)
		//	rc = wolfSSL_CTX_use_PrivateKey_buffer(client->tls.ctx,
		//		(const byte*)device_priv_key, (long)XSTRLEN(device_priv_key),
		//		WOLFSSL_FILETYPE_PEM);
	}

	PRINTF("MQTT TLS Setup (%d)", rc);

	return rc;
}
/* callback indicates a network error occurred */
static int mqtt_disconnect_cb(MqttClient* client, int error_code, void* ctx)
{
	mv_mqtt_instance_impl_t* impl = (mv_mqtt_instance_impl_t*)client->ctx;
	(void)client;
	(void)ctx;
	PRINTF("Network Error Callback: %s (error %d)",
		MqttClient_ReturnCodeToString(error_code), error_code);
	if (impl->mqtt_conn_status_t != nullptr)
		impl->mqtt_conn_status_t(ctx, error_code);
	//重连
	int timer = 0;
	while (1)
	{
		uint32_t result = mv_mqtt_connect(impl);
		if (result == 1|| result == 0) {
			std::cout << "cause: connect_success" << std::endl;
			if (impl->mqtt_conn_status_t != nullptr)
				impl->mqtt_conn_status_t(ctx, result);
			break;
		}
		else
		{
			printf("Failed to connect, return code %d\n", result);
			if (impl->mqtt_conn_status_t != nullptr)
				impl->mqtt_conn_status_t(ctx, result);
		}
		sleep(10);
	}
	return 0;
}

int mqtt_message_cb(MqttClient* client, MqttMessage* msg,
	byte msg_new, byte msg_done)
{
	byte buf[PRINT_BUFFER_SIZE + 1];
	word32 len;
	mv_mqtt_instance_impl_t* impl = (mv_mqtt_instance_impl_t*)client->ctx;
	//(void)mqttCtx;

	wm_SemLock(&mtLock);
	if (msg_new) {
		/* Determine min size to dump */
		//len = msg->topic_name_len;
		//if (len > PRINT_BUFFER_SIZE) {
		//	len = PRINT_BUFFER_SIZE;
		//}
		//XMEMCPY(buf, msg->topic_name, len);
		//buf[len] = '\0'; /* Make sure its null terminated */

		/* Print incoming message */
		PRINTF("MQTT Message: Topic %s, Qos %d, Id %d, Len %u, %u, %u",
			msg->topic_name, msg->qos, msg->packet_id, msg->total_len, msg->buffer_len, msg->buffer_pos);
		auto iterator = impl->handle_function.find(msg->topic_name);
		if (iterator == impl->handle_function.end()) {
			if (impl->mqtt_msg_handler_t == nullptr) {
				printf("[WARNING] %s handle_function is not exists\n, message is: %s\n", msg->topic_name, msg->buffer);
			}
			else {
				impl->mqtt_msg_handler_t(impl, msg->topic_name, reinterpret_cast<const char*>(msg->buffer));
			}
		}
		else {
			(*iterator).second(impl, msg->topic_name, reinterpret_cast<const char*>(msg->buffer));
		}
	}

	/* Print message payload */
	//len = msg->buffer_len;
	//if (len > PRINT_BUFFER_SIZE) {
	//	len = PRINT_BUFFER_SIZE;
	//}
	//XMEMCPY(buf, msg->buffer, len);
	//buf[len] = '\0'; /* Make sure its null terminated */
	//PRINTF("Payload (%d - %d) printing %d bytes:" LINE_END "%s",
	//	msg->buffer_pos, msg->buffer_pos + msg->buffer_len, len, buf);
	if (msg_done) {
		PRINTF("MQTT Message: Done");
	}
	wm_SemUnlock(&mtLock);

	/* Return negative to terminate publish processing */
	return MQTT_CODE_SUCCESS;
}
int mv_wmq_mqtt_init(mv_mqtt_instance_impl_t* impl) {
	/* setup tx/rx buffers */
	impl->mqttCtx.tx_buf = (byte*)WOLFMQTT_MALLOC(MAX_BUFFER_SIZE);
	impl->mqttCtx.rx_buf = (byte*)WOLFMQTT_MALLOC(MAX_BUFFER_SIZE);
	/* Initialize MqttClient structure */
	int rc = MqttClient_Init(&impl->mqttCtx.client, &impl->mqttCtx.net, mqtt_message_cb,
		impl->mqttCtx.tx_buf, MAX_BUFFER_SIZE, impl->mqttCtx.rx_buf, MAX_BUFFER_SIZE,
		impl->mqttCtx.cmd_timeout_ms);
	if (rc == MQTT_CODE_CONTINUE) {
		return rc;
	}
	PRINTF("MQTT Init: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}
	impl->mqttCtx.client.ctx = impl;

#ifdef WOLFMQTT_V5
	/* AWS broker only supports v3.1.1 client */
	impl->mqttCtx.client.protocol_level = MQTT_CONNECT_PROTOCOL_LEVEL_5;
#endif

#ifdef WOLFMQTT_DISCONNECT_CB
	/* setup disconnect callback */
	rc = MqttClient_SetDisconnectCallback(&impl->mqttCtx.client,
		mqtt_disconnect_cb, impl);
	if (rc != MQTT_CODE_SUCCESS) {
		mv_wmq_client_cleanup(impl);
	}
#endif
	return MQTT_CODE_SUCCESS;
}

int mv_wmq_tcp_conn(mv_mqtt_instance_impl_t* impl) {
	/* Connect to broker */
	int rc = MqttClient_NetConnect(&impl->mqttCtx.client, impl->mqttCtx.host, impl->mqttCtx.port,
		DEFAULT_CON_TIMEOUT_MS, impl->mqttCtx.use_tls, mqtt_aws_tls_cb);
	if (rc == MQTT_CODE_CONTINUE) {
		return rc;
	}
	PRINTF("MQTT Socket Connect: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}

	/* Build connect packet */
	XMEMSET(&impl->mqttCtx.connect, 0, sizeof(MqttConnect));
	impl->mqttCtx.connect.keep_alive_sec = impl->mqttCtx.keep_alive_sec;
	impl->mqttCtx.connect.clean_session = impl->mqttCtx.clean_session;
	impl->mqttCtx.connect.client_id = impl->mqttCtx.client_id;

	/* Last will and testament sent by broker to subscribers
		of topic when broker connection is lost */
	//XMEMSET(&mqttCtx->lwt_msg, 0, sizeof(mqttCtx->lwt_msg));
	//mqttCtx->connect.lwt_msg = &mqttCtx->lwt_msg;
	//mqttCtx->connect.enable_lwt = mqttCtx->enable_lwt;
	//if (mqttCtx->enable_lwt) {
	//	/* Send client id in LWT payload */
	//	mqttCtx->lwt_msg.qos = mqttCtx->qos;
	//	mqttCtx->lwt_msg.retain = 0;
	//	mqttCtx->lwt_msg.topic_name = AWSIOT_PUBLISH_TOPIC"lwt";
	//	mqttCtx->lwt_msg.buffer = (byte*)mqttCtx->client_id;
	//	mqttCtx->lwt_msg.total_len = (word16)XSTRLEN(mqttCtx->client_id);
	//}
	/* Optional authentication */
	impl->mqttCtx.connect.username = impl->mqttCtx.username;
	impl->mqttCtx.connect.password = impl->mqttCtx.password;
	return MQTT_CODE_SUCCESS;
}
int mv_wmq_mqtt_conn(mv_mqtt_instance_impl_t* impl) {
	/* Send Connect and wait for Connect Ack */
	int rc = MqttClient_Connect(&impl->mqttCtx.client, &impl->mqttCtx.connect);
	if (rc == MQTT_CODE_CONTINUE) {
		return rc;
	}
	PRINTF("MQTT Connect: Proto (%s), %s (%d)",
		MqttClient_GetProtocolVersionString(&impl->mqttCtx.client),
		MqttClient_ReturnCodeToString(rc), rc);
	if (rc != MQTT_CODE_SUCCESS) {
		mv_wmq_disconn(impl);
		return rc;
	}
	/* Validate Connect Ack info */
	PRINTF("MQTT Connect Ack: Return Code %u, Session Present %d",
		impl->mqttCtx.connect.ack.return_code,
		(impl->mqttCtx.connect.ack.flags &
			MQTT_CONNECT_ACK_FLAG_SESSION_PRESENT) ?
		1 : 0
	);
	return MQTT_CODE_SUCCESS;
}

int mv_wmq_sub(mv_mqtt_instance_impl_t* impl, const char * topic, int qos) {
	int rc = MQTT_CODE_SUCCESS;
	uint16_t i;
	word32 startSec = 0;

	/* Build list of topics */
	XMEMSET(&impl->mqttCtx.subscribe, 0, sizeof(MqttSubscribe));
	impl->mqttCtx.topics[0].topic_filter = topic;
	impl->mqttCtx.topics[0].qos = (MqttQoS)qos;

#ifdef WOLFMQTT_V5
	if (impl->mqttCtx.subId_not_avail != 1) {
		/* Subscription Identifier */
		MqttProp* prop;
		prop = MqttClient_PropsAdd(&impl->mqttCtx.subscribe.props);
		prop->type = MQTT_PROP_SUBSCRIPTION_ID;
		prop->data_int = DEFAULT_SUB_ID;
	}
#endif

	/* Subscribe Topic */
	impl->mqttCtx.subscribe.packet_id = mqtt_get_packetid_threadsafe();
	impl->mqttCtx.subscribe.topic_count =
		sizeof(impl->mqttCtx.topics) / sizeof(MqttTopic);
	impl->mqttCtx.subscribe.topics = impl->mqttCtx.topics;

	do {
		rc = MqttClient_Subscribe(&impl->mqttCtx.client, &impl->mqttCtx.subscribe);
		rc = check_response(impl, rc, &startSec, MQTT_PACKET_TYPE_SUBSCRIBE);
	} while (rc == MQTT_CODE_CONTINUE);
	if (rc != MQTT_CODE_SUCCESS) {
		MqttClient_CancelMessage(&impl->mqttCtx.client,
			(MqttObject*)&impl->mqttCtx.subscribe);
	}

	PRINTF("MQTT Subscribe: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);

	if (rc == MQTT_CODE_SUCCESS) {
		/* show subscribe results */
		MqttTopic* topic = &impl->mqttCtx.subscribe.topics[0];
		PRINTF("  Topic %s, Qos %u, Return Code %u",
			topic->topic_filter,
			topic->qos, topic->return_code);
		rc = topic->return_code;
	}

#ifdef WOLFMQTT_V5
	if (impl->mqttCtx.subscribe.props != NULL) {
		MqttClient_PropsFree(impl->mqttCtx.subscribe.props);
	}
#endif
	return rc;
}

int init_mv_wmq_sub(mv_mqtt_instance_impl_t* impl) {
	int rc = MQTT_CODE_SUCCESS;
	uint16_t i;
	word32 startSec = 0;

	/* Build list of topics */
	XMEMSET(&impl->mqttCtx.subscribe, 0, sizeof(MqttSubscribe));
	impl->mqttCtx.topics[0].topic_filter = impl->topic_device_register_reply.c_str();
	impl->mqttCtx.topics[0].qos = MQTT_QOS_1;
	impl->mqttCtx.topics[1].topic_filter = impl->topic_property_get.c_str();
	impl->mqttCtx.topics[1].qos = MQTT_QOS_1;
	impl->mqttCtx.topics[2].topic_filter = impl->topic_property_set.c_str();
	impl->mqttCtx.topics[2].qos = MQTT_QOS_1;
	impl->mqttCtx.topics[3].topic_filter = impl->topic_tsevent_post_reply.c_str();
	impl->mqttCtx.topics[3].qos = MQTT_QOS_1;
	impl->mqttCtx.topics[4].topic_filter = impl->topic_event_post_reply.c_str();
	impl->mqttCtx.topics[4].qos = MQTT_QOS_1;
	impl->mqttCtx.topics[5].topic_filter = impl->topic_action_exec.c_str();
	impl->mqttCtx.topics[5].qos = MQTT_QOS_1;
#ifdef WOLFMQTT_V5
	if (impl->mqttCtx.subId_not_avail != 1) {
		/* Subscription Identifier */
		MqttProp* prop;
		prop = MqttClient_PropsAdd(&impl->mqttCtx.subscribe.props);
		prop->type = MQTT_PROP_SUBSCRIPTION_ID;
		prop->data_int = DEFAULT_SUB_ID;
	}
#endif

	/* Subscribe Topic */
	impl->mqttCtx.subscribe.packet_id = mqtt_get_packetid_threadsafe();
	impl->mqttCtx.subscribe.topic_count = 6;
		//sizeof(impl->mqttCtx.topics) / sizeof(MqttTopic);
	impl->mqttCtx.subscribe.topics = impl->mqttCtx.topics;

	do {
		rc = MqttClient_Subscribe(&impl->mqttCtx.client, &impl->mqttCtx.subscribe);
		rc = check_response(impl, rc, &startSec, MQTT_PACKET_TYPE_SUBSCRIBE);
	} while (rc == MQTT_CODE_CONTINUE);
	if (rc != MQTT_CODE_SUCCESS) {
		MqttClient_CancelMessage(&impl->mqttCtx.client,
			(MqttObject*)&impl->mqttCtx.subscribe);
	}

	PRINTF("MQTT Subscribe: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);

	if (rc == MQTT_CODE_SUCCESS) {
		/* show subscribe results */
		for (i = 0; i < impl->mqttCtx.subscribe.topic_count; i++) {
			MqttTopic* topic = &impl->mqttCtx.subscribe.topics[i];
			PRINTF("  Topic %s, Qos %u, Return Code %u",
				topic->topic_filter,
				topic->qos, topic->return_code);
			if (topic->return_code != 0)
				rc = topic->return_code;
		}
	}

#ifdef WOLFMQTT_V5
	if (impl->mqttCtx.subscribe.props != NULL) {
		MqttClient_PropsFree(impl->mqttCtx.subscribe.props);
	}
#endif
	return rc;
}
int mv_wmq_unsub(mv_mqtt_instance_impl_t* impl, const char* topic) {
	int rc= MQTT_CODE_SUCCESS;
	word32 startSec = 0;

	/* Build list of topics */
	XMEMSET(&impl->mqttCtx.subscribe, 0, sizeof(MqttSubscribe));
	impl->mqttCtx.topics[0].topic_filter = topic;
	impl->mqttCtx.topics[0].qos = impl->mqttCtx.qos;

	/* Unsubscribe Topics */
	XMEMSET(&impl->mqttCtx.unsubscribe, 0, sizeof(MqttUnsubscribe));
	impl->mqttCtx.unsubscribe.packet_id = mqtt_get_packetid_threadsafe();
	impl->mqttCtx.unsubscribe.topic_count =
		sizeof(impl->mqttCtx.topics) / sizeof(MqttTopic);
	impl->mqttCtx.unsubscribe.topics = impl->mqttCtx.topics;

	/* Unsubscribe Topics */
	do {
		rc = MqttClient_Unsubscribe(&impl->mqttCtx.client, &impl->mqttCtx.unsubscribe);
		rc = check_response(impl, rc, &startSec, MQTT_PACKET_TYPE_UNSUBSCRIBE);
	} while (rc == MQTT_CODE_CONTINUE);
	if (rc != MQTT_CODE_SUCCESS) {
		MqttClient_CancelMessage(&impl->mqttCtx.client,
			(MqttObject*)&impl->mqttCtx.unsubscribe);
	}

	PRINTF("MQTT Unsubscribe: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);

	return rc;
}

int mv_wmq_pub(mv_mqtt_instance_impl_t* impl, const char* topic, const char* message, int qos) {
	int rc;
	MqttPublish publish;
	word32 startSec = 0;

	/* Publish Topic */
	XMEMSET(&publish, 0, sizeof(MqttPublish));
	publish.retain = 0;
	publish.qos = (MqttQoS)qos;
	publish.duplicate = 0;
	publish.topic_name = topic;
	publish.packet_id = mqtt_get_packetid_threadsafe();
	publish.buffer = (byte*)message;
	publish.total_len = (word16)XSTRLEN(message);

	do {
		rc = MqttClient_Publish(&impl->mqttCtx.client, &publish);
		rc = check_response(impl, rc, &startSec, MQTT_PACKET_TYPE_PUBLISH);
	} while (rc == MQTT_CODE_CONTINUE);
	if (rc != MQTT_CODE_SUCCESS) {
		MqttClient_CancelMessage(&impl->mqttCtx.client, (MqttObject*)&publish);
	}

	PRINTF("MQTT Publish: Topic %s, %s (%d)",
		publish.topic_name,
		MqttClient_ReturnCodeToString(rc), rc);
	return rc;
}
int mv_wmq_wait_msg(mv_mqtt_instance_impl_t* impl) {
	int rc = MQTT_CODE_SUCCESS;
	do {
		/* check for test mode or stop */
		if (mStopRead || impl->mqttCtx.test_mode) {
			rc = MQTT_CODE_SUCCESS;
			PRINTF("MQTT Exiting...");
			break;
		}

		/* Try and read packet */
		rc = MqttClient_WaitMessage(&impl->mqttCtx.client, impl->mqttCtx.cmd_timeout_ms);

#ifdef WOLFMQTT_NONBLOCK
		/* Track elapsed time with no activity and trigger timeout */
		rc = mqtt_check_timeout(rc, &impl->mqttCtx.start_sec,
			impl->mqttCtx.cmd_timeout_ms / 1000);
#endif

		/* check return code */
		if (rc == MQTT_CODE_CONTINUE) {
			return rc;
		}
#ifdef WOLFMQTT_ENABLE_STDIN_CAP
		else if (rc == MQTT_CODE_STDIN_WAKE) {
			/* Get data from STDIO */
			XMEMSET(impl->mqttCtx.rx_buf, 0, MAX_BUFFER_SIZE);
			if (XFGETS((char*)impl->mqttCtx.rx_buf, MAX_BUFFER_SIZE - 1, stdin) != NULL) {
				/* rc = (int)XSTRLEN((char*)mqttCtx->rx_buf); */

				/* Publish Topic */
				XSNPRINTF((char*)impl->mqttCtx.app_ctx, AWSIOT_PUBLISH_MSG_SZ,
					"{\"state\":{\"reported\":{\"msg\":\"%s\"}}}",
					impl->mqttCtx.rx_buf);
				impl->mqttCtx.stat = WMQ_PUB;
				XMEMSET(&impl->mqttCtx.publish, 0, sizeof(MqttPublish));
				impl->mqttCtx.publish.retain = 0;
				impl->mqttCtx.publish.qos = impl->mqttCtx.qos;
				impl->mqttCtx.publish.duplicate = 0;
				impl->mqttCtx.publish.topic_name = AWSIOT_PUBLISH_TOPIC;
				impl->mqttCtx.publish.packet_id = mqtt_get_packetid();
				impl->mqttCtx.publish.buffer = (byte*)impl->mqttCtx.app_ctx;
				impl->mqttCtx.publish.total_len = (word32)XSTRLEN((char*)impl->mqttCtx.app_ctx);
				rc = MqttClient_Publish(&impl->mqttCtx.client, &impl->mqttCtx.publish);
				PRINTF("MQTT Publish: Topic %s, %s (%d)",
					impl->mqttCtx.publish.topic_name,
					MqttClient_ReturnCodeToString(rc), rc);
			}
		}
#endif
		else if (rc == MQTT_CODE_ERROR_TIMEOUT) {
			/* Keep Alive */
			PRINTF("Keep-alive timeout, sending ping");

			rc = MqttClient_Ping_ex(&impl->mqttCtx.client, &impl->mqttCtx.ping);
			if (rc == MQTT_CODE_CONTINUE) {
				return rc;
			}
			else if (rc != MQTT_CODE_SUCCESS) {
				PRINTF("MQTT Ping Keep Alive Error: %s (%d)",
					MqttClient_ReturnCodeToString(rc), rc);
				break;
			}
		}
		else if (rc != MQTT_CODE_SUCCESS) {
			/* There was an error */
			PRINTF("MQTT Message Wait: %s (%d)",
				MqttClient_ReturnCodeToString(rc), rc);
			break;
		}
	} while (1);

	/* Check for error */
	if (rc != MQTT_CODE_SUCCESS) {
		return mv_wmq_disconn(impl);
	}
	return MQTT_CODE_SUCCESS;
}

int mv_wmq_ping(mv_mqtt_instance_impl_t* impl) {
	int rc = MQTT_CODE_SUCCESS;

	/* Disconnect */
	rc = MqttClient_Disconnect_ex(&impl->mqttCtx.client,
		&impl->mqttCtx.disconnect);
	if (rc == MQTT_CODE_CONTINUE) {
		return rc;
	}
	PRINTF("MQTT Disconnect: %s (%d)",
		MqttClient_ReturnCodeToString(rc), rc);
	if (rc != MQTT_CODE_SUCCESS) {
		return mv_wmq_disconn(impl);
	}
	return MQTT_CODE_SUCCESS;
}
std::string mv_mqtt_get_version() {
	return PROJECT_VERSION;
}
mv_mqtt_instance_t mv_mqtt_create_instance() {
	std::cout << "mqtt_version: " << mv_mqtt_get_version() << std::endl;
	mv_mqtt_instance_impl_t* impl = new mv_mqtt_instance_impl_t();
	return impl;
}

std::string no_nullptr_string_from_ini(CSimpleIni& ini, const char* section, const char* key) {
	const char* temp_string = nullptr;
	if (section == nullptr) {
		return "";
	}
	temp_string = ini.GetValue(section, key);
	if (temp_string == nullptr) {
		return "";
	}
	else {
		return temp_string;
	}
}
/**
 * @brief  string类型字符串分割函数
 * @param  str(string): 待分割字符串
 * @param  separator(string): 分割符号
 * @return result: 分割结果vetor容器
 */
std::string getFirstElement(std::string str, const std::string& separator)
{
	std::vector<std::string> result;
	int cutAt;
	while ((cutAt = str.find_first_of(separator)) != str.npos)
	{
		if (cutAt > 0)
		{
			result.push_back(str.substr(0, cutAt));
		}
		str = str.substr(cutAt + 1);
	}
	if (str.length() > 0)
	{
		result.push_back(str);
	}
	int num = -1;
	num = 0;
	if (num >= 0)
		return std::move(result[num]);

	return "";
}

/**
 * @brief  string类型字符串分割函数
 * @param  str(string): 待分割字符串
 * @param  separator(string): 分割符号
 * @return result: 分割结果vetor容器
 */
std::string getLastElement(std::string str, const std::string& separator)
{
	std::vector<std::string> result;
	int cutAt;
	while ((cutAt = str.find_first_of(separator)) != str.npos)
	{
		if (cutAt > 0)
		{
			result.push_back(str.substr(0, cutAt));
		}
		str = str.substr(cutAt + 1);
	}
	if (str.length() > 0)
	{
		result.push_back(str);
	}

	int num = -1;
	num = result.size() - 1;
	if (num >= 0)
		return std::move(result[num]);

	return "";
}
int load_config(mv_mqtt_instance_impl_t* impl) {
	CSimpleIni ini;
	SI_Error rc = ini.LoadFile(impl->config_path.c_str());
	if (rc < 0) {
		printf("Cannot open %s\n", impl->config_path.c_str());
		return -1;
	}

	std::list<CSimpleIni::Entry> sections;
	ini.GetAllSections(sections);

	if (sections.empty()) {
		printf("%s all is empty \n", impl->config_path.c_str());
		return -1;
	}
	impl->product_key = no_nullptr_string_from_ini(ini, "Secret", "ProductKey");
	impl->device_name = no_nullptr_string_from_ini(ini, "Secret", "DeviceName");
	impl->client_id = no_nullptr_string_from_ini(ini, "Secret", "ClientId");
	impl->device_secret = no_nullptr_string_from_ini(ini, "Secret", "DeviceSecret");
	impl->register_time = no_nullptr_string_from_ini(ini, "Secret", "RegisterTime");
	impl->product_secret = no_nullptr_string_from_ini(ini, "Secret", "ProductSecret");
	impl->app_id = no_nullptr_string_from_ini(ini, "Secret", "AppId");
	impl->sn = no_nullptr_string_from_ini(ini, "Secret", "SN");

	impl->CApath = no_nullptr_string_from_ini(ini, "SSLOptions", "CApath");

	impl->username = no_nullptr_string_from_ini(ini, "ConnectOptions", "Username");
	impl->password = no_nullptr_string_from_ini(ini, "ConnectOptions", "Password"); 
	impl->broker_address = no_nullptr_string_from_ini(ini, "ConnectOptions", "BrokerAddress");
	impl->broker_port = no_nullptr_string_from_ini(ini, "ConnectOptions", "BrokerPort");
	impl->cloud_address = no_nullptr_string_from_ini(ini, "ConnectOptions", "CloudAddress");

	std::cout << "SN:                   " << impl->sn << std::endl;
	std::cout << "product_key:          " << impl->product_key << std::endl;
	std::cout << "device_name:          " << impl->device_name << std::endl;
	std::cout << "client_id:            " << impl->client_id << std::endl;
	std::cout << "device_secret:        " << impl->device_secret << std::endl;
	std::cout << "register_time:        " << impl->register_time << std::endl;
	std::cout << "product_secret:       " << impl->product_secret << std::endl;
	std::cout << "CApath:               " << impl->CApath << std::endl;
	std::cout << "username:             " << impl->username << std::endl;
	std::cout << "password:             " << impl->password << std::endl;
	std::cout << "broker_address:       " << impl->broker_address << std::endl;
	std::cout << "broker_port:          " << impl->broker_port << std::endl;
	std::cout << "app_id:               " << impl->app_id << std::endl;
	std::cout << "cloud_address:        " << impl->cloud_address << std::endl;

	if (impl->product_key == "" || impl->device_name == "" || impl->client_id == "" || /*impl->key_store == "" || impl->trust_store == "" || impl->private_key_password == "" ||*/impl->CApath == "" || impl->username == "" || impl->password == "" || impl->broker_address == "" || impl->broker_port == "" || impl->app_id == "" || impl->cloud_address == "") {
		printf(R"([ERROR] impl->product_key == "" || impl->device_name == "" || impl->client_id == "" || impl->CApath == "" || impl->username == "" || impl->password == "" || impl->broker_address == "" || impl->broker_port == "" || impl->app_id == "" || impl->cloud_address = "")");
		return -1;
	}

	return 0;
}

void initMqttCtx(mv_mqtt_instance_impl_t* impl) {
	//处理ip
	//std::string targ = getLastElement(impl->cloud_address, "\/");
	//std::string ip   = getFirstElement(impl->broker_address, ":");
	//std::string port = getLastElement(impl->broker_port, ":");
	char pubMsg[AWSIOT_PUBLISH_MSG_SZ] = { 0 };
	/* init defaults */
	mqtt_init_ctx(&impl->mqttCtx);
	impl->mqttCtx.app_name = "maxvision";
	//impl->mqttCtx.host = AWSIOT_HOST;
	impl->mqttCtx.host = impl->broker_address.c_str();
	impl->mqttCtx.port = (word16)std::stoi(impl->broker_port);
	//       mqttCtx.port = AWSIOT_PORT;
	//impl->mqttCtx.qos = AWSIOT_QOS;
	impl->mqttCtx.keep_alive_sec = AWSIOT_KEEP_ALIVE_SEC;
	impl->mqttCtx.client_id = impl->client_id.c_str();
	//impl->mqttCtx.topic_name = AWSIOT_SUBSCRIBE_TOPIC;
	impl->mqttCtx.cmd_timeout_ms = AWSIOT_CMD_TIMEOUT_MS;
	impl->mqttCtx.use_tls = 3;
	impl->mqttCtx.app_ctx = pubMsg;
	/* init defaults */
	impl->mqttCtx.username = impl->username.c_str();
	impl->mqttCtx.password = impl->password.c_str();
}

void init_topic(mv_mqtt_instance_impl_t* impl)
{
#define KEY_NAME impl->product_key + "/" + impl->device_name
	//设备注册
	impl->topic_device_register = "/maxvision/sys/device/register";
	impl->topic_device_register_reply = "/maxvision/sys/" + KEY_NAME + "/device/register_reply";
	//属性设置
	impl->topic_property_set = "/maxvision/sys/" + KEY_NAME + "/thing/property/set";
	impl->topic_property_set_reply = "/maxvision/sys/" + KEY_NAME + "/thing/property/set_reply";
	//属性获取
	impl->topic_property_get = "/maxvision/sys/" + KEY_NAME + "/thing/property/get";
	impl->topic_property_get_reply = "/maxvision/sys/" + KEY_NAME + "/thing/property/get_reply";
	//动作
	impl->topic_action_exec = "/maxvision/sys/" + KEY_NAME + "/thing/action/+/exec";
	impl->topic_action_exec_reply = "/maxvision/sys/" + KEY_NAME + "/thing/action/+/exec_reply";
	//时序事件
	impl->topic_tsevent_post = "/maxvision/sys/" + KEY_NAME + "/thing/tsevent/+/post";
	impl->topic_tsevent_post_reply = "/maxvision/sys/" + KEY_NAME + "/thing/tsevent/+/post_reply";
	//非时序事件
	impl->topic_event_post = "/maxvision/sys/" + KEY_NAME + "/thing/event/+/post";
	impl->topic_event_post_reply = "/maxvision/sys/" + KEY_NAME + "/thing/event/+/post_reply";
}
// 设备注册
// topic: thing/sub/register
int register_message_handle(void* context, const char* topic, const char* received_message) {
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(context);
	std::cout << "register_message_handle start" << std::endl;
	std::cout << "--------------------------------------" << std::endl;
	std::cout << topic << std::endl;
	std::cout << received_message << std::endl;
	std::cout << "register_message_handle end" << std::endl;
	return 0;
}
// 绑定事件和处理函数
void init_function_map(mv_mqtt_instance_impl_t* impl) {

#define REGISTER_FUNCTION(KEY,VALUE) \
    impl->handle_function[KEY] = VALUE; printf("register function: %s ==> %s\n", KEY.c_str(), " " #VALUE);
	REGISTER_FUNCTION(impl->topic_device_register_reply, register_message_handle);
#undef REGISTER_FUNCTION
}
int mv_mqtt_init_instance(mv_mqtt_instance_t instance, const std::string& config_path, mqtt_msg_handler_callback_t handle) {
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(instance);
	int rc = MQTT_CODE_SUCCESS;

	//这个步骤会通过加载配置文件修改
	impl->config_path = config_path;
	if (load_config(impl) != 0)
		return -1;

	//初始化MqttCtx
	initMqttCtx(impl);
	init_topic(impl);
	init_function_map(impl);
	//注册消息回调
	mv_set_mqtt_msg_handler_callback(impl, handle);
	return rc;
}
int mv_mqtt_connect(mv_mqtt_instance_t instance) {
	keep = false;
	int rc = MQTT_CODE_SUCCESS;
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(instance);
	// 初始化net模块
	rc = mv_wmq_net_init(impl);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}
	// 初始化mqtt模块
	rc = mv_wmq_mqtt_init(impl);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}
	// tcp连接
	rc = mv_wmq_tcp_conn(impl);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}
	// mqtt连接
	rc = mv_wmq_mqtt_conn(impl);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}
	keep = true;
	rc = init_mv_wmq_sub(impl);
	if (rc != MQTT_CODE_SUCCESS) {
		return rc;
	}
	return rc;
}
int mv_mqtt_subscribe(mv_mqtt_instance_t instance, const std::string& topic, int qos) {
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(instance);
	return mv_wmq_sub(impl,topic.c_str(),qos);
}

int mv_mqtt_publish(mv_mqtt_instance_t instance, const std::string& topic, const std::string& message, int qos) {
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(instance);
	return mv_wmq_pub(impl, topic.c_str(), message.c_str(), qos);
}
int mv_mqtt_unsubscribe(mv_mqtt_instance_t instance, const std::string& topic) {
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(instance);
	return mv_wmq_unsub(impl, topic.c_str());
}
int mv_mqtt_disconnect(mv_mqtt_instance_t instance, int timeout) {
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(instance);
	return mv_wmq_disconn(impl);
}

int mv_mqtt_close(mv_mqtt_instance_t instance) {
	mv_mqtt_instance_impl_t* impl = reinterpret_cast<decltype(impl)>(instance);
	mv_wmq_client_cleanup(impl);
	delete impl;
	impl = nullptr;
	return 0;
}
//-------------------------------------------------------------------
// 客户端向代理服务器主动发送数据
//-------------------------------------------------------------------

void mv_set_mqtt_msg_handler_callback(mv_mqtt_instance_t instance, mqtt_msg_handler_callback_t callback) {
	reinterpret_cast<mv_mqtt_instance_impl_t*>(instance)->mqtt_msg_handler_t = callback;
}
void mv_set_mqtt_connection_status_callback(mv_mqtt_instance_t instance, mqtt_connection_status_callback_t callback) {
	reinterpret_cast<mv_mqtt_instance_impl_t*>(instance)->mqtt_conn_status_t = callback;
}
//-------------------------------------------------------------------

